package com.baishancloud.log.quality

import cn.hutool.core.exceptions.ExceptionUtil
import cn.hutool.json.{JSONObject, JSONUtil}
import com.baishancloud.log.common.entity.{CountryEntity, DomainEntity, IspEntity, ProvinceEntity}
import com.baishancloud.log.common.util.{ExternalDataFetchUtil, StringUtil, TimeUtil}
import com.baishancloud.log.format.java.util.IPLocator
import com.baishancloud.log.quality.pojo.{StarRocksWeb, Tags, WebLog}
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.configuration.Configuration
import org.slf4j.{Logger, LoggerFactory}

import java.lang
import java.util.{Timer, TimerTask}
import scala.collection.mutable

/**
 * 对输入的web端日志进行对象转换<br>
 * 输出为为二元组(WebLog, StarRocksWeb)，第一个元素为计算比率的WebLog，第二个值为写入doris明细数据的StarRocksWeb对象
 *
 * @author ziqiang.wang 
 * @date 2021/12/7 16:18
 */
class WebLogParse extends RichMapFunction[String, (WebLog, StarRocksWeb)] {

  private val LOGGER: Logger = LoggerFactory.getLogger(classOf[WebLogParse])

  // 域名id信息表
  private val domainInfo: mutable.Map[String, DomainEntity] = mutable.Map[String, DomainEntity]()
  // 国家id信息表
  private val countryInfo: mutable.Map[String, CountryEntity] = mutable.Map[String, CountryEntity]()
  // 省份id信息表
  private val provinceInfo: mutable.Map[String, ProvinceEntity] = mutable.Map[String, ProvinceEntity]()
  // 运营商id信息表
  private val ispInfo: mutable.Map[String, IspEntity] = mutable.Map[String, IspEntity]()

  private var subTaskIndex: Int = -1


  //允许迟到数据最长时间
  private var allowOldest: Long = System.currentTimeMillis() - 24 * 3600 * 1000


  /**
   * 初始化缓存数据，增加定时器，定时更新缓存数据
   */
  override def open(parameters: Configuration): Unit = {
    subTaskIndex = getRuntimeContext.getIndexOfThisSubtask
    ExternalDataFetchUtil.domainInfoFetch(domainInfo)
    ExternalDataFetchUtil.countryInfoFetch(countryInfo)
    ExternalDataFetchUtil.provinceInfoFetch(provinceInfo)
    ExternalDataFetchUtil.ispInfoFetch(ispInfo)
    LOGGER.info(s"domainInfo${subTaskIndex}初始化完成，domainId表size为：${domainInfo.size}")
    LOGGER.info(s"countryInfo${subTaskIndex}初始化完成，countryId表size为：${countryInfo.size}")
    LOGGER.info(s"provinceInfo${subTaskIndex}初始化完成，provinceId表size为：${provinceInfo.size}")
    LOGGER.info(s"ispInfo${subTaskIndex}初始化完成，ispId为：${ispInfo.size}")
    val timer: Timer = new Timer("缓存数据更新", true)
    //增加定时器，每半小时获取一次最新缓存数据
    timer.scheduleAtFixedRate(new TimerTask {
      override def run(): Unit = {
        ExternalDataFetchUtil.domainInfoFetch(domainInfo)
        ExternalDataFetchUtil.countryInfoFetch(countryInfo)
        ExternalDataFetchUtil.provinceInfoFetch(provinceInfo)
        ExternalDataFetchUtil.ispInfoFetch(ispInfo)
        allowOldest = System.currentTimeMillis() - 24 * 3600 * 1000
      }
    }, 1800 * 1000, 1800 * 1000)
  }


  override def map(value: String): (WebLog, StarRocksWeb) = {
    try {
      val jsonObj: JSONObject = JSONUtil.parseObj(value)
      val cTime: lang.Long = jsonObj.getLong("ctime")
      //过滤迟到数据
      if (cTime < allowOldest) return (null, null)
      val domain: String = StringUtil.subString(jsonObj.getJSONObject("params").getStr("sHost"), 200)
      val valueJsonObj: JSONObject = {
        try {
          jsonObj.getJSONObject("value")
        } catch {
          case _: Exception => null
        }
      }

      //ip信息
      val ipInfo = IPLocator.lookup(jsonObj.getStr("ip"))
      var isp = ipInfo.getIsp.split("/")(0)
      if (isp.equals("other")) isp = "其他"
      if (isp.contains("广电")) {
        isp = "广电"
      } else if (isp.contains("华数")) {
        isp = "华数"
      }
      val country = StringUtil.subString(if (ipInfo.getCountry.equals("other")) "其他" else ipInfo.getCountry, 33)
      var province = StringUtil.subString(if (ipInfo.getProvince.equals("other")) "其他" else ipInfo.getProvince, 33)
      if ((country.equals("中国") || country.startsWith("CHINA")) && province.equals("其他")) province = "中国其他地区"

      (
        WebLog(
          Tags(
            duration,
            null,
            businessLive,
            sourceWeb,
            domain,
            province,
            isp,
            null,
            null,
            ""
          ),
          cTime,
          jsonObj.getInt("category"),
          jsonObj.getInt("code"),
          jsonObj.getStr("uvid"),
          jsonObj.getStr("value")
        ),
        StarRocksWeb(
          "live",
          TimeUtil.timeAtLong(cTime, 60 * 1000) / 1000,
          TimeUtil.timeAtString(cTime, 60 * 1000),
          domainInfo.getOrElse(domain, DomainEntity.DEFAULT).domainId,
          domain,
          jsonObj.getByte("category"),
          jsonObj.getByte("code"),
          ispInfo.getOrElse(isp, IspEntity.DEFAULT).id,
          isp,
          countryInfo.getOrElse(country, CountryEntity.DEFAULT).id,
          country,
          provinceInfo.getOrElse(province, ProvinceEntity.DEFAULT).id,
          province,
          jsonObj.getStr("uvid"),
          if (valueJsonObj == null) jsonObj.getInt("value") else null,
          if (valueJsonObj != null) valueJsonObj.getInt("windowSize") else null,
          if (valueJsonObj != null) valueJsonObj.getInt("count") else null,
          if (valueJsonObj != null) valueJsonObj.getInt("duration") else null,
          if (valueJsonObj != null) valueJsonObj.getInt("playTime") else null,
          if (valueJsonObj != null) valueJsonObj.getInt("playedTime") else null,
          if (valueJsonObj != null) valueJsonObj.getInt("receivedBytes") else null
        )
      )
    } catch {
      case e: Exception => {
        LOGGER.error(s"web live-json内容提取出错，数据为：\n${value}\n错误信息为：" + ExceptionUtil.stacktraceToString(e, 1000))
        (null, null)
      }
    }
  }
}
